package queues

import (
	"context"
	"sync"
	"time"

	"github.com/gogf/gf/v2/os/glog"
)

// 队列配置
type QueueConfig struct {
	MaxQueueSize int           // 最大队列长度
	BatchSize    int           // 批量处理大小
	MaxInterval  time.Duration // 最大处理间隔
}

// 内存队列实现
type BatchQueue struct {
	config    QueueConfig
	messages  []interface{}
	mu        sync.Mutex
	stopChan  chan struct{}
	processor func([]interface{}) // 处理函数
	running   bool                // 新增运行状态标志
}

var (
	queues     = make(map[string]*BatchQueue)
	queuesLock sync.Mutex
)

// 创建或获取队列实例
func GetOrCreateQueue(queueType string, config QueueConfig, processor func([]interface{})) *BatchQueue {
	queuesLock.Lock()
	defer queuesLock.Unlock()

	if q, exists := queues[queueType]; exists {
		return q
	}

	q := NewBatchQueue(config, processor)
	queues[queueType] = q
	return q
}

func GetQueue(queueType string) (*BatchQueue, bool) {
	a, b := queues[queueType]
	return a, b
}

// 修改Start方法添加运行状态检查
func (q *BatchQueue) Start() {
	q.mu.Lock()
	if q.running {
		q.mu.Unlock()
		return
	}
	q.running = true
	q.mu.Unlock()

	ticker := time.NewTicker(q.config.MaxInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			q.processBatch()
		case <-q.stopChan:
			q.mu.Lock()
			q.running = false
			q.mu.Unlock()
			return
		default:
			q.mu.Lock()
			if len(q.messages) >= q.config.BatchSize {
				q.mu.Unlock()
				q.processBatch()
			} else {
				q.mu.Unlock()
				time.Sleep(100 * time.Millisecond) // 避免CPU空转
			}
		}
	}
}

// 创建新队列
func NewBatchQueue(config QueueConfig, processor func([]interface{})) *BatchQueue {
	return &BatchQueue{
		config:    config,
		messages:  make([]interface{}, 0, config.MaxQueueSize),
		stopChan:  make(chan struct{}),
		processor: processor,
	}
}

// 投递数据到队列
func (q *BatchQueue) Push(data interface{}) bool {
	q.mu.Lock()
	defer q.mu.Unlock()

	if len(q.messages) >= q.config.MaxQueueSize {
		glog.Warning(context.Background(), "队列已满，丢弃消息")
		return false
	}

	q.messages = append(q.messages, data)
	return true
}

// 停止队列
func (q *BatchQueue) Stop() {
	close(q.stopChan)
}

// 处理批次数据
func (q *BatchQueue) processBatch() {
	q.mu.Lock()
	if len(q.messages) == 0 {
		q.mu.Unlock()
		return
	}

	batchSize := q.config.BatchSize
	if len(q.messages) < batchSize {
		batchSize = len(q.messages)
	}

	batch := make([]interface{}, batchSize)
	copy(batch, q.messages[:batchSize])
	q.messages = q.messages[batchSize:]
	q.mu.Unlock()

	// 调用处理函数
	q.processor(batch)
}
